/*
    异步工作器：
        采用双缓冲区
        业务线程将数据保存到输入缓冲区
        如果 输出缓冲区为空 或者 输入缓冲区已满，则交换缓冲区，由异步线程继续将输出缓冲区的内容写入到磁盘
        但两个缓冲区都为空时，不发生交换，异步线程则挂起等待数据
*/

#pragma once
#include "buffer.hpp"
#include <memory>
#include <thread>
#include <mutex>
#include <atomic>
#include <functional>
#include <condition_variable>

namespace XKL
{
    using Functor = std::function<void(Buffer&)>;

    enum AsyncType
    {
        ASYNC_SAFE,
        ASYNC_UNSAFE
    };

    class AsyncLooper
    {
    public:
        using ptr = std::shared_ptr<AsyncLooper>;
        AsyncLooper(const Functor& callback, AsyncType looper_type = ASYNC_SAFE) // callback 写入函数
            : _looper_type(looper_type), _stop(false), _callback(callback), _th(std::thread(&AsyncLooper::ThreadEntry, this))
        {}

        ~AsyncLooper()
        {
            Stop();
        }
        
        void Stop()
        {
            _stop = true;
            _cond_con.notify_all(); // 唤醒所有线程处理完缓冲区内剩余的数据
            _th.join();             // 等待线程退出
        }

        void Push(const char* data, size_t len) // 将多种情况交给工作器处理 buffer直接扩容就行了
        {
            // 1. 无限扩容（buffer内部自动扩容） -- 非安全 2. 固定大小 -- 当缓冲区数据满了就直接挂起阻塞
            std::unique_lock<std::mutex> lock(_mtx);
            if(_looper_type == ASYNC_SAFE)
                _cond_pro.wait(lock, [&](){ return len <= _pro_buff.WriteAbleSize(); });
            _pro_buff.push(data, len);
            _cond_con.notify_all();
        }
    private:
        // 线程入口函数--对消费缓冲区中的数据进行处理，处理完后，初始化缓冲区
        void ThreadEntry()
        {
            while(true)
            {
                {
                    std::unique_lock<std::mutex> lock(_mtx);
                    // 等缓冲区真正完全清空后才能退出
                    if(_stop == true && _pro_buff.Empty())
                        break;
                    _cond_con.wait(lock, [&](){ return _stop || !_pro_buff.Empty(); });
                    _con_buff.Swap(_pro_buff);

                    if(_looper_type == ASYNC_SAFE)
                        _cond_pro.notify_all();
                }

                _callback(_con_buff);
                _con_buff.Rest();
            }
        }
    private:
        Buffer _pro_buff; // 生产缓冲区
        Buffer _con_buff; // 消费缓冲区
        std::mutex _mtx;
        std::thread _th;
        Functor _callback;
        AsyncType _looper_type;
        std::atomic<bool> _stop;       // 工作器停止标志 
        std::condition_variable _cond_pro;
        std::condition_variable _cond_con;
    };
}